package com.ermao.mymes.cron.service.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import com.ermao.mymes.cron.pending.CrowdBatchTaskPending;
import com.ermao.mymes.cron.service.TaskHandler;
import com.ermao.mymes.cron.utils.ReadFileUtils;
import com.ermao.mymes.cron.vo.CrowdInfoVo;
import com.ermao.mymes.support.dao.MessageTemplateDao;
import com.ermao.mymes.support.domain.MessageTemplate;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.util.HashMap;

/**
 * 获取原型，CrowdBatchTaskPending，并且将本次处理的内容
 * 一行行地添加到阻塞队列中
 *
 * @author Ermao
 * Date: 2022/5/18 9:33
 */
@Service
@Slf4j
public class TaskHandlerImpl implements TaskHandler {
	@Autowired
	private MessageTemplateDao messageTemplateDao;

	@Autowired
	private ApplicationContext context;

	@Override
	@Async
	public void handle(Long messageTemplateId) {
		log.info("TaskHandler handle:{}", Thread.currentThread().getName());

		MessageTemplate messageTemplate = messageTemplateDao.findById(messageTemplateId).get();
		if (messageTemplate == null || StrUtil.isBlank(messageTemplate.getCronCrowdPath())) {
			log.error("TaskHandler#handle crowdPath empty! messageTemplateId:{}", messageTemplateId);
			return;
		}

		CrowdBatchTaskPending crowdBatchTaskPending = context.getBean(CrowdBatchTaskPending.class);
		// 读取文件得到每一行记录给到队列做lazy batch处理
		ReadFileUtils.getCsvRow(messageTemplate.getCronCrowdPath(), row -> {
			if (CollUtil.isEmpty(row.getFieldMap())
					|| StrUtil.isBlank(row.getFieldMap().get(ReadFileUtils.RECEIVER_KEY))) {
				return;
			}
			HashMap<String, String> params = ReadFileUtils.getParamFromLine(row.getFieldMap());
			CrowdInfoVo crowdInfoVo = CrowdInfoVo.builder().receiver(row.getFieldMap().get(ReadFileUtils.RECEIVER_KEY))
					.params(params).messageTemplateId(messageTemplateId).build();
			crowdBatchTaskPending.pending(crowdInfoVo);
		});

	}
}
